package com.boot.study.window

import com.boot.study.api.{MySensorSource, SensorReading}
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object WindowTest {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //    env.getConfig.setAutoWatermarkInterval(500) // 全局设置watermark
    //    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 时间语义 事件时间

    // 自定义source
    val dataStream: DataStream[SensorReading] = env.addSource(new MySensorSource())
      //      .assignAscendingTimestamps(_.timeStamp * 1000) // 升序数据提取时间搓
      // 最大乱序程度为3秒
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(element: SensorReading): Long = element.timeStamp * 1000
      })

    // 定义侧输出流
    val lateTag: OutputTag[(String, Double, Long)] = new OutputTag[(String, Double, Long)]("late");

    // 每15秒统计一次窗口内各传感器温度最小值
    val resultStream = dataStream.map(data => (data.id, data.temperature, data.timeStamp))
      .keyBy(_._1) // 按照二元组的第一个元素分组
      //      .window(TumblingEventTimeWindows.of(Time.seconds(15))) // 滚动时间窗口
      //      .window(SlidingEventTimeWindows.of(Time.seconds(15), Time.seconds(3))) // 滑动时间窗口
      //      .window(EventTimeSessionWindows.withGap(Time.seconds(15))) // 会话窗口
      .timeWindow(Time.seconds(1)) // 滚动窗口
      .allowedLateness(Time.minutes(1)) // 允许处理迟到数据
      .sideOutputLateData(lateTag) // 侧输出流
      //      .countWindow(10) // 滚动计数窗口
      //      .minBy(1)
      .reduce((cur, news) => (cur._1, cur._2.min(news._2), news._3))

    resultStream.print("result")
    // 获取侧输出流
    resultStream.getSideOutput(lateTag).print("late")
    env.execute("window test")
  }
}

// 自定义函数
class MyReduce extends ReduceFunction[SensorReading] {
  override def reduce(value1: SensorReading, value2: SensorReading): SensorReading = {
    SensorReading(value1.id, value2.timeStamp, value1.temperature.min(value2.temperature))
  }
}
